//! @Author: DengLibin
//! @Date: Create in 2023-10-24 10:05:14
//! @Description: 文件异步io操作
use encoding::all::GBK;
use encoding::{DecoderTrap, Encoding};
use std::ops::Deref;
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::{
    future::Future,
    pin::Pin,
    sync::{atomic::AtomicU32, Arc},
};
use tokio::fs::{self, File};
use tokio::sync::Mutex;
use tokio::{
    fs::read_dir,
    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt},
    sync::{mpsc, Semaphore},
    task,
};

/// @Author: DengLibin
/// @Date: Create in 2025-03-13 16:16:04
/// @Description: 删除文件
pub async fn remove_file(path: &str) -> Result<(), Box<dyn std::error::Error>> {
    let r = tokio::fs::remove_file(path).await?; //删除文件
    Ok(r)
}

/// @Author: DengLibin
/// @Date: Create in 2025-03-13 16:16:58
/// @Description: 删除文件夹及内容
pub async fn remove_dir_and_contents(path: &str) -> Result<(), std::io::Error> {
    fs::remove_dir_all(path).await
}

/// @Author: DengLibin
/// @Date: Create in 2025-03-13 16:11:53
/// @Description: 创建文件夹
pub async fn create_dir_if_no_exists(dir_path: &str) -> Result<(), Box<dyn std::error::Error>> {
    // 检查路径是否存在
    if !fs::metadata(dir_path).await.is_ok() {
        // 如果路径不存在，则创建目录
        let _ = fs::create_dir_all(dir_path).await?;
    }
    Ok(())
}

/// @Author: DengLibin
/// @Date: Create in 2025-03-13 15:59:00
/// @Description: 是否存在
pub async fn exists(path: &str) -> Result<bool, Box<dyn std::error::Error>> {
    let r = fs::try_exists(path).await?;
    Ok(r)
}

/// @Author: DengLibin
/// @Date: Create in 2023-10-24 10:06:09
/// @Description: 按行读取文本文件 示例
pub async fn read_by_line<F: FnMut(String)->bool>(
    path: &str,
    mut func: F,
) -> Result<(), Box<dyn std::error::Error>> {
    let file = tokio::fs::File::open(path).await?;
    // 将file转换为BufReader
    let mut buf_reader = tokio::io::BufReader::with_capacity(1024 * 1024, file).lines();
    // 每次读取一行
    while let Some(line) = buf_reader.next_line().await.unwrap() {
        // 注意lines()中的行是不带结尾换行符的，因此使用println!()而不是print!()
        if !func(line) {
            break;
        }
    }
    Ok(())
}
/// @Author: DengLibin
/// @Date: Create in 2023-10-24 10:06:09
/// @Description: 按行读取文本文件 示例
pub async fn read_by_line2<F: FnMut(String)>(
    path: &str,
    mut func: F,
) -> Result<(), Box<dyn std::error::Error>> {
    let file = tokio::fs::File::open(path).await?;
    let mut buf_reader = tokio::io::BufReader::with_capacity(1024 * 1024, file);
    let mut buf = String::new();

    return loop {
        match buf_reader.read_line(&mut buf).await {
            Err(e) => break Err(e.into()),
            // 遇到了文件结尾，即EOF
            Ok(0) => break Ok(()),
            Ok(_n) => {
                func(buf);
                buf = String::new();
            }
        }
    };
}
/// @Author: DengLibin
/// @Date: Create in 2023-10-24 10:21:52
/// @Description: 拷贝文件
pub async fn copy_file(path: &str, to_path: &str) -> Result<u64, Box<dyn std::error::Error>> {
    let r = tokio::fs::copy(path, to_path).await?;
    Ok(r)
}
/// @Author: DengLibin
/// @Date: Create in 2023-10-24 10:21:52
/// @Description: 拷贝文件
/// @progress_func: 进度百分比回调
pub async fn copy_file2<F: Fn(u8)>(
    path: &str,
    to_path: &str,
    progress_func: F,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut file = tokio::fs::File::open(path).await?;
    let metadata = file.metadata().await?;
    let len = metadata.len();
    let mut to_file = tokio::fs::File::create(to_path).await?;
    let mut buffer: [u8; 10240] = [0; 10240];
    let mut progress: u64 = 0;
    return loop {
        match file.read(&mut buffer).await {
            Err(e) => break Err(e.into()),
            // 遇到了文件结尾，即EOF
            Ok(0) => break Ok(()),
            Ok(n) => {
                progress += n as u64;
                progress_func((progress * 100 / len) as u8);

                to_file.write_all(&buffer[0..n]).await?;
                to_file.flush().await?;
            }
        }
    };
}

/// @Author: DengLibin
/// @Date: Create in 2025-03-18 16:49:58
/// @Description: 写入内容
pub async fn write_str(text: &str, out_path: &str) -> Result<(), Box<dyn std::error::Error>> {
    let parent_o = Path::new(out_path).parent();
    if let Some(parent) = parent_o {
        create_dir_if_no_exists(parent.to_str().unwrap()).await?;
    }
    let mut file = File::create(out_path).await?;
    file.write_all(text.as_bytes()).await?;
    file.flush().await?;
    Ok(())
}

/// @Author: DengLibin
/// @Date: Create in 2024-07-26 09:40:33
/// @Description: 删除文件获取文件夹
pub async fn delete_file_or_dir(file: &str) -> Result<(), Box<dyn std::error::Error>> {
    let metadata = tokio::fs::metadata(file).await?;
    if metadata.is_file() {
        tokio::fs::remove_file(file).await?;
    } else if metadata.is_dir() {
        //读取子文件(文件夹)
        let mut read_dir = tokio::fs::read_dir(file).await?;
        while let Some(entry) = read_dir.next_entry().await? {
            let path = entry.path();
            if let Some(p) = path.to_str() {
                delete_file_or_dir_recursive(p.into()).await?; //递归
            }
        }
        //删除空文件夹
        tokio::fs::remove_dir(file).await?;
    } else if metadata.is_symlink() {
        //符号链接
        let target_path = tokio::fs::read_link(file).await?;
        let path = target_path.to_str();
        if let Some(real_path) = path {
            delete_file_or_dir_recursive(real_path.into()).await?;
        }
    }

    Ok(())
}

/// @Author: DengLibin
/// @Date: Create in 2025-01-15 10:10:26
/// @Description: 统计文件夹下文件数量
pub async fn count_sub_file(
    dir: String,
    counter: Arc<AtomicU32>,
) -> Result<Vec<String>, Box<dyn std::error::Error>> {
    let semaphore = Arc::new(Semaphore::new(400)); // 限制并发数
                                                   //无界多生产者单消费者通道
    let (tx, mut rx) = mpsc::unbounded_channel();

    let active_tasks = Arc::new(AtomicUsize::new(1)); // 活跃任务计数器
    // 启动初始任务
    tx.send(dir)?;


    let mut handles = Vec::new();
    let mut paths = Vec::new();
    // 消费队列
    while let Some(path) = rx.recv().await {
        if path.is_empty() {
            //收到空表示结束
            break;
        }
        let tx_clone = tx.clone();
        let semaphore_clone = semaphore.clone();
        let counter_clone = Arc::clone(&counter);
        let active_tasks_clone = Arc::clone(&active_tasks);
        //获取执行权限
        let permit = semaphore_clone.acquire_owned().await?;

        let h = task::spawn(async move {
            let mut files = Vec::new();
            // 遍历目录
            if let Ok(mut entries) = read_dir(&path).await {
                while let Ok(Some(entry)) = entries.next_entry().await {
                    if let Ok(file_type) = entry.file_type().await {
                       

                        // 如果是目录，加入队列
                        if file_type.is_dir() {
                            let sub_dir = entry.path().to_string_lossy().into_owned();
                            let r = tx_clone.send(sub_dir.clone());
                            if let Err(e) = r {
                                println!("发送文件夹失败:{}, {}", sub_dir, e);
                            } else {
                                //发送了一个新任务
                                active_tasks_clone.fetch_add(1, Ordering::SeqCst);
                            }
                        } else if file_type.is_file() {
                            // 计数
                            let c = counter_clone.fetch_add(1, Ordering::SeqCst);
                            if c % 1000 == 0 {
                                log::info!("count:{}", c);
                            }
                           
                            //文件
                            files.push(entry.path().to_string_lossy().into_owned());
                        }
                    }
                }
            }

            //当前目录消费完成
            let i = active_tasks_clone.fetch_sub(1, Ordering::SeqCst);

            if 1 == i {
                //最后一个了

                let _ = tx_clone.send("".into()); //发送一个空
            }

            drop(permit); // 释放信号量
            files
        });
        handles.push(h);
    }
    for h in handles {
        let mut sub_paths = h.await?;
        paths.append(&mut sub_paths);
    }


    Ok(paths)
}

/// @Author: DengLibin
/// @Date: Create in 2024-07-26 09:59:53
/// @Description: 递归包装
/// 异步函数在 Rust 中通过返回一个实现了 Future 特征的类型来表示。
/// Rust 编译器需要知道函数返回值的大小（在编译时），但异步函数的返回值大小通常不是固定的，
/// 因为它们在等待时可能会产生不同大小的中间状态。
/// Rust 编译器需要在编译时知道每个类型的大小。如果类型的大小在编译时不确定（例如，递归异步函数）
/// ，则需要使用某种方式将其大小固定下来。
/// Box::pin 是一种将类型的大小固定下来的方法。它将一个动态大小的类型（如递归异步函数的返回值）转换为一个固定大小的类型。
/// 通过将递归调用包装在 Pin<Box<dyn Future<Output = io::Result<()>>>> 中，我们可以将递归异步函数的返回值固定下来，
/// 从而解决编译器无法确定大小的问题。
fn delete_file_or_dir_recursive(
    path: String,
) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error>>>>> {
    Box::pin(async move { delete_file_or_dir(path.as_str()).await })
}

/// @Author: DengLibin
/// @Date: Create in 2025-03-18 16:27:45
/// @Description: 读取文本文件内容
pub async fn read_all_txt(file_path: &str) -> Result<String, Box<dyn std::error::Error>> {
    // 打开文件
    let mut file = File::open(file_path).await?;

    // 创建一个字符串缓冲区来存储文件内容
    let mut buffer = Vec::new();

    // 异步读取文件内容到缓冲区
    file.read_to_end(&mut buffer).await?;

    let is_utf8 = is_utf8(&buffer);
    if is_utf8 {
        return Ok(String::from_utf8_lossy(&buffer).into());
    }

    // 将字节缓冲区转换为指定编码的字符串
    let decoded_r = GBK.decode(&buffer, DecoderTrap::Strict);
    let content = match decoded_r {
        Ok(content) => content,
        Err(_) => "".into(),
    };

    Ok(content)
}

/// @Author: DengLibin
/// @Date: Create in 2024-07-31 14:31:13
/// @Description: 是否是utf8编码
fn is_utf8(buffer: &[u8]) -> bool {
    match std::str::from_utf8(buffer) {
        Ok(_) => true,
        Err(_) => false,
    }
}
